/*
 * server.cpp
 *
 * 	Ex: 5
 *
 *  Created on: Jun 6, 2012
 *      Author: Ariel Ayash
 */

#include "server.h"

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <fcntl.h>
#include <cstdio>
#include <sys/time.h>
#include <cassert>
#include <string>
#include <algorithm>
#include <cerrno>
#include <cstring> // for memset
#include <iostream>


using namespace std;


#define LOG_FILE_NAME "twitServer.log"
#define MIN_PORT_NUM (1)
#define MAX_PORT_NUM (0xffff)
#define MAX_WAITING_CLIENTS (FD_SETSIZE)
#define EXIT_COMMAND "EXIT"


void Server::print_err_msg( const char *err_message, bool to_errno )
{
	_log_file << err_message << endl;

	if ( to_errno )
	{
		perror( err_message );
	}
	else
	{
		cerr << err_message << endl;
	}

}

// open a server socket on the given valid port
result_t Server::open_socket( uint32_t port )
{
	struct sockaddr_in my_addr;
	int sys_res;
	result_t result = OPERATION_SUCCESS;


	// creating an Internet TCP socket
	_hsock = socket( AF_INET, SOCK_STREAM, 0 );
	if ( -1 == _hsock )
	{
		ERR_PRINT("Error: while opening socket!\n");
		print_err_msg( "Error: while opening socket", true );
		return ERROR_SYSTEM_CALL;
	}

	// make the socket port reusable
	int yes = 1;
	sys_res = setsockopt ( _hsock, SOL_SOCKET, SO_REUSEADDR, & yes, sizeof(int) );
	if ( sys_res )
	{
		ERR_PRINT("Error: setsockopt() failed!\n");
		print_err_msg( "Error: while making socket's port reusable", true );
		result = ERROR_SYSTEM_CALL;
		goto END;
	}

	// assign address for the server
	my_addr.sin_family = AF_INET;
	my_addr.sin_port = htons( port );
	my_addr.sin_addr.s_addr = htonl( INADDR_ANY );
	memset( my_addr.sin_zero, 0, 8 );

	// connect the socket to the address + port
	sys_res = bind( _hsock, (struct sockaddr *) & my_addr, sizeof(struct sockaddr) );
	if ( sys_res )
	{
		ERR_PRINT("Error: bind() failed!\n");
		print_err_msg( "Error: while binding socket", true );
		result = ERROR_SYSTEM_CALL;
		goto END;
	}

	// create the accept queue
	sys_res = listen( _hsock, MAX_WAITING_CLIENTS );
	if ( sys_res )
	{
		ERR_PRINT("Error: listen() failed!\n");
		print_err_msg( "Error: while calling listen on socket", true );
		result = ERROR_SYSTEM_CALL;
		goto END;
	}

	// make the socket non-blocked
	sys_res = fcntl ( _hsock, F_SETFL, O_NONBLOCK );
	if ( sys_res )
	{
		ERR_PRINT("Error: fcntl() failed!\n");
		print_err_msg( "Error: while making socket non-blocked" );
		result = ERROR_SYSTEM_CALL;
		goto END;
	}

	DBG_PRINT("server socket %d was created on port: %u\n",
			  _hsock,
			  port);

END:
	if( OPERATION_SUCCESS != result )
	{
		close( _hsock );
		_hsock = INVALID_SOCKET;
	}


	return result;
}

Server::Server( int port, result_t &result ) :
	_hsock ( INVALID_SOCKET ),
	_is_running(false),
	_max_file_desc_num(1)	// maximal is stdin + 1
{
	result = OPERATION_SUCCESS;

	_log_file.open( LOG_FILE_NAME , ios_base::out | ios_base::app );
	if ( _log_file.fail() )
	{
		ERR_PRINT("Error: while opening log file!\n");
		print_err_msg( "Error: while creating a log file" );
		result = ERROR_IO;

		return;
	}

	result = open_socket( port );
	if ( OPERATION_SUCCESS != result )
	{
		goto END;
	}

	// clean the read set
	FD_ZERO( & _in_set );


	/*
	 *  set in the input set the STDIN in order to get messages from terminal and
	 *  the server socket in order to accept new clients
	 */
	FD_SET( STDIN_FILENO, & _in_set );
	FD_SET( _hsock, & _in_set );


END:

	if ( OPERATION_SUCCESS != result )
	{
		_log_file.close();
	}
	else
	{
		_max_file_desc_num = _hsock + 1;
	}
}

Server::~Server()
{
	DBG_PRINT("Server's resources were deleted\n");
}

void Server::server_start()
{
	std::string line;
	fd_set read_set;
	fd_set write_set;
	fd_set exception_set;
	uint remaining_fd;
	int sys_res;

	list< deferred_msg_t >::iterator it;

	_is_running = true;

	while ( _is_running || ! _deferred_msg.empty() )
	{
		// prepare a copy of the sets that should be monitored
		memcpy( & read_set, & _in_set, sizeof(read_set) );
		memcpy( & exception_set, & _in_set, sizeof(read_set) );
		FD_CLR( STDIN_FILENO, & exception_set );
		FD_CLR( _hsock, & exception_set );

		// set all the messages that are waiting to be sent
		FD_ZERO( & write_set );
		for ( it = _deferred_msg.begin(); it != _deferred_msg.end(); ++it )
		{
			FD_SET( it->socket, & write_set );
		}

		// wait for a file descriptor to be ready
		sys_res = select( _max_file_desc_num, & read_set,  & write_set, & exception_set ,NULL );
		if ( -1 == sys_res )
		{
			print_err_msg( "Error: while using select() on the socket" , true );

			_is_running = false;

			break;
		}
		else
		{
			remaining_fd = sys_res;
		}

		assert( remaining_fd ); // will fail if there are no waiting sockets

		// check for new clients
		if ( FD_ISSET( _hsock , & read_set )  )
		{
			accept_new_client();

			--remaining_fd;
		}

		// check for EXIT command
		if ( FD_ISSET( STDIN_FILENO, & read_set ) )
		{
			cin >> line;

			if ( EXIT_COMMAND == line )
			{
				_is_running = false;

				// clean the read set
				FD_ZERO( & _in_set );

				DBG_PRINT("EXIT command was received\n");
			}

#ifdef DEBUG
			else if ( "DUMP" == line )
			{
				// dump all the data that find in the server
				map< uint, ClientContext* >::iterator dbg_it;
				for ( dbg_it = _clients.begin(); dbg_it != _clients.end(); ++dbg_it )
				{
					cout<< "socket: " << dbg_it->first << "\t name: "
							<< dbg_it->second->get_client_name();


					cout << "\tfollowers: ";
					vector<uint> followers_sockets;

					dbg_it->second->get_client_followers_socket( followers_sockets );

					vector<uint>::iterator it;
					for ( it = followers_sockets.begin(); it != followers_sockets.end(); ++it )
					{
						cout<< *it << " ";
					}

					cout<< endl;

				}

				cout << endl << endl << "\ttotal clients: " << _clients.size() << endl;
				cout << "\ttotal names: " << _names.size() << endl;
			}

#endif

			--remaining_fd;
		}

		// close all the clients that had problem with their socket
		if ( remaining_fd )
		{
			check_for_exception(  & exception_set, remaining_fd );
		}

		// send deferred messages
		if ( remaining_fd && _is_running ) // there are more file descriptors to handle
		{
			send_deferred_msgs( & write_set, remaining_fd );
		}

		// read messages from clients
		if ( remaining_fd && _is_running ) // there are more file descriptors to handle
		{
			get_msgs( & read_set, remaining_fd );
		}

	}

	// release server's resources
	close_server();
}

void Server::accept_new_client()
{
	int csock;
	struct sockaddr_in addr;
	ClientContext *client = NULL;

	uint addrlen = sizeof( addr );

	csock = accept( _hsock, ( struct sockaddr *) & addr, & addrlen  );
	if ( csock < 0 )
	{
		ERR_PRINT("Error: accept() failed!\n");
		print_err_msg( "Error: while trying to accept a new client", true );
		return;
	}

	DBG_PRINT("Accepting a new client with socket: %d\n", csock);

	assert( _clients.find( _hsock ) == _clients.end() );

	client = new ClientContext();

	// add the client to the known clients
	_clients[csock] = client;

	// increase the maximum number of file descriptors that known
	_max_file_desc_num = max( _max_file_desc_num, uint( csock + 1) );

	/*
	 * add the new client to the set of known clients so we would get 
	 * his messages
	 */
	FD_SET( csock, & _in_set );
}

void Server::send_deferred_msgs( fd_set *ready, uint & num_to_handle)
{
	list< deferred_msg_t >::iterator def_it;
	list< deferred_msg_t >::iterator tmp_it; // will be used when removing element

	deferred_msg_t def_msg;

	uint msg_size;
	void *start_from;
	int bytecount;

	for ( def_it = _deferred_msg.begin(); def_it != _deferred_msg.end(); ++def_it )
	{

		if ( ! num_to_handle )
		{
			// there are no more messages to handle
			break;
		}

		def_msg = *def_it;

		// iterate over all the clients and check whose socket ready for transmit
		if ( ! FD_ISSET( def_msg.socket , ready )  )
		{
			// the client's socket is not ready for transmit
			continue;
		}

		--num_to_handle;

		if ( def_msg.first_time )
		{
			msg_size = MSG_METADATA_LENGTH + def_msg.msg->header.payload_length;

			// change byte order
			def_msg.msg->header.type = (messages_type_t)htonl(def_msg.msg->header.type);
			def_msg.msg->header.payload_length = htonl(def_msg.msg->header.payload_length);
			def_msg.msg->header.status = (response_status_t)htonl(def_msg.msg->header.status);

			start_from = (void *)def_msg.msg;
		}
		else
		{
			msg_size = ntohl(def_msg.msg->header.payload_length) - def_msg.already_sent;

			start_from = & def_msg.msg->payload[ def_msg.already_sent ];
		}

		bytecount = send( def_msg.socket, start_from , msg_size, 0 );

		if ( bytecount > 0 &&  bytecount < (int) msg_size  )
		{
			// we need another iteration for this message
			def_msg.already_sent += bytecount;

			def_msg.first_time = false;

			DBG_PRINT("Sending a message in more then one iteration\n");

			continue;
		}
		if( bytecount <= 0 )
		{
			// in case of error we don't retry sending the message

			print_err_msg( "Error: while sending message", true );
		}

		free( def_msg.msg );

		// remove the message that was sent
		def_it = _deferred_msg.erase( def_it );

	}
}

void Server::check_for_exception(  fd_set *exception_set, uint & num_to_handle )
{
	map< uint, ClientContext* >::iterator it;

	vector<uint> clients_to_remove;

	for ( it = _clients.begin(); it != _clients.end(); it++ )
	{
		if ( ! FD_ISSET( it->first , exception_set )  )
		{
			// the current client didn't send message
			continue;
		}

		--num_to_handle;

		close_connection_with_client( it->first );

		string err_msg("Error: exception closing the port");
		print_err_msg( err_msg.c_str() );
		clients_to_remove.push_back( it->first );

	}

	// remove the clients that should be removed
	vector< uint >::iterator rm_it;
	for ( rm_it = clients_to_remove.begin(); rm_it != clients_to_remove.end(); ++rm_it )
	{
		_clients.erase( *rm_it );
	}

}

void Server::get_msgs( fd_set *waiting, uint &num_to_handle)
{
	map< uint, ClientContext* >::iterator it;

	vector<uint> clients_to_remove;
	result_t result;

	int bytecount;
	in_msg_t message;

	message.payload = NULL;

	uint payload_current_length = 0;


	for ( it = _clients.begin(); it != _clients.end(); it++ )
	{
		if ( ! num_to_handle )
		{
			break;
		}

		if ( ! FD_ISSET( it->first , waiting )  )
		{
			// the current client didn't send message
			continue;
		}

		--num_to_handle;

		// read the header
		bytecount = recv( it->first, & message, MSG_METADATA_LENGTH, 0 );

		if ( ! bytecount )
		{
			/*
			 * Empty message - can be only occur when the client closed the socket without sending
			 * disconnect message.
			 */
			ERR_PRINT("client_name: %s socket: %u closed the socket without DISCONNECT!\n",
					  it->second->get_client_name(),
					  it->first);
			
			close_connection_with_client( it->first );

			clients_to_remove.push_back( it->first );

			continue;
		}

		if( MSG_METADATA_LENGTH != bytecount )
		{
			ERR_PRINT("Error: while receiving data from client_name: %s socket: %u\n",
					  it->second->get_client_name(),
					  it->first);

			print_err_msg( "Error: while receiving data", true );

			close_connection_with_client( it->first );

			clients_to_remove.push_back( it->first );

			continue;
		}

		// decode the type
		message.header.type = (messages_type_t)ntohl( message.header.type );
		if (  message.header.type <= MESSAGE_TYPE_MIN_NUM
				|| message.header.type >= MESSAGE_TYPE_MAX_NUM )
		{
			// unknown type

			ERR_PRINT("Unknown type of message was received by socket %u\n", it->first);

			close_connection_with_client( it->first );

			clients_to_remove.push_back( it->first );

			continue;
		}


		// decode the length
		message.header.payload_length = ntohl( message.header.payload_length );

		// check if there is a payload to the message
		if ( message.header.payload_length )
		{
			// get the buffer

			// allocate more space if needed
			if ( payload_current_length < message.header.payload_length )
			{
				if ( NULL != message.payload )
				{
					free( message.payload );
					message.payload  = NULL;

				}

				payload_current_length = message.header.payload_length;
				message.payload = (char *) malloc( message.header.payload_length );

				if ( NULL == message.payload )
				{
					/*
					 * In case there is no memory we kill the client.
					 * There are many ways to handle this case. we decided to kill the client
					 * for security purposes - make sure the client doesn't try to cause denial
					 * of service.
					 */
					ERR_PRINT("Too long message was received from socket %u - closing him\n", it->first);

					close_connection_with_client( it->first );

					clients_to_remove.push_back( it->first );

					continue;
				}
			}

			// get the rest of the message
			bytecount = recv( it->first, message.payload, message.header.payload_length, 0 );
			if(  bytecount < 0 || message.header.payload_length != (uint32_t)bytecount )
			{

				ERR_PRINT("Error: receiving message payload from client_name: %s socket: %u\n",
						  it->second->get_client_name(),
						  it->first);

				print_err_msg( "Error: while receiving data", true );

				close_connection_with_client( it->first );

				clients_to_remove.push_back( it->first );

				continue;
			}
		}

		// parse the message
		result = parse_message( it->first, & message );
		if ( REMOVE_CLIENT  == result )
		{
			clients_to_remove.push_back( it->first );
		}
	}

	if ( NULL != message.payload )
	{
		free( message.payload );
		message.payload = NULL;
	}

	// remove the clients that should be removed
	vector< uint >::iterator rm_it;
	for ( rm_it = clients_to_remove.begin(); rm_it != clients_to_remove.end(); ++rm_it )
	{
		_clients.erase( *rm_it );
	}
}

void Server::close_server()
{
	DBG_PRINT("Closing server\n");


	// delete any deferred messages
	deferred_msg_t dm;
	list< deferred_msg_t >:: iterator msg_it;
	for ( msg_it = _deferred_msg.begin(); msg_it != _deferred_msg.end(); ++msg_it )
	{
		dm = *msg_it;

		if ( NULL != dm.msg )
		{
			free( dm.msg );
			dm.msg = NULL;
		}
	}

	_deferred_msg.clear();

	// close all clients' open sockets
	map< uint, ClientContext* >:: iterator it;
	for ( it = _clients.begin(); it != _clients.end(); ++it )
	{
		close_connection_with_client( it->first );
	}

	// close server's socket
	if ( INVALID_SOCKET != _hsock )
	{
		sleep(1);
		_log_file << "Closing the server!" << endl;
		if ( shutdown( _hsock, SHUT_RDWR ) || close( _hsock ) )
		{
			ERR_PRINT("Error: close() for server socket failed!\n");
			print_err_msg( "Error: closing server socket failed", true );
		}

		_hsock = INVALID_SOCKET;
	}

	// close log file handle
	if ( _log_file.is_open() )
	{
		_log_file.close();
		if ( _log_file.fail() )
		{
			ERR_PRINT("Error: close() on log file failed!\n");
			print_err_msg( "Error: closing log file failed", true );
		}
	}
}


//	------------------------------------------------------------------------------>
//						End of Server implementation
//	------------------------------------------------------------------------------>

int main(int argc, char** argv)
{
	Server *server = NULL;
	result_t result;

	// validate arguments
	if ( 2 != argc)
	{
		cerr << "Usage: " << argv[0] <<  " <port num>"<< endl;
		return EXIT_FAILURE;
	}

	int port = atoi( argv[1] );
	if ( ! port || port < MIN_PORT_NUM || port > MAX_PORT_NUM)
	{
		cerr << "Invalid port number" << endl;
		return EXIT_FAILURE;
	}

	server = new Server(port, result);
	if ( OPERATION_SUCCESS != result )
	{
		return EXIT_FAILURE;
	}

	server->server_start();

	delete server;

	return EXIT_SUCCESS;
}
